package com.atguigu.order.pay.detect;

import com.atguigu.order.pay.detect.bean.OrderEvent;
import com.atguigu.order.pay.detect.bean.OrderResult;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.net.URL;

public class OrderTimeoutWithoutCep {
    private static final OutputTag<OrderResult> orderTimeoutTag = new OutputTag<OrderResult>("pay timeout") {
    };

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        URL resource = OrderPayTimeout.class.getResource("/OrderLog.csv");
        DataStreamSource<String> dataStreamSource = env.readTextFile(resource.getPath());

        SingleOutputStreamOperator<OrderEvent> orderEventStream = dataStreamSource.map(line -> {
            String[] fields = line.split(",");
            return new OrderEvent(new Long(fields[0]), fields[1], fields[2], new Long(fields[3]));
        }).assignTimestampsAndWatermarks(new AscendingTimestampExtractor<OrderEvent>() {
            @Override
            public long extractAscendingTimestamp(OrderEvent element) {
                return element.getTimestamp() * 1000L;
            }
        });

        //自定义处理函数，主流输出正常支付订单事件，侧输出流输出超时报警事件
        SingleOutputStreamOperator<OrderResult> resultStream = orderEventStream.keyBy(OrderEvent::getOrderId)
                .process(new OrderPayMatchDetect());

        resultStream.print("paid normally");

        resultStream.getSideOutput(orderTimeoutTag).print("pay timeout");

        env.execute("order detect pay without cep");
    }

    public static class OrderPayMatchDetect extends KeyedProcessFunction<Long, OrderEvent, OrderResult> {
        //保存之前订单是否已经来过
        ValueState<Boolean> isPaidState;
        ValueState<Boolean> isCreatedState;
        //定时器时间戳
        ValueState<Long> tsTimestampState;

        @Override
        public void open(Configuration parameters) throws Exception {
            isPaidState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("is-paid", Boolean.class, false));
            isCreatedState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("is-created", Boolean.class, false));
            tsTimestampState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("ts", Long.class));
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<OrderResult> out) throws Exception {
            //定时器触发，一定有一个事件没来
            if (isPaidState.value()) {
                //如果pay来了，说明create没来
                ctx.output(orderTimeoutTag, new OrderResult(ctx.getCurrentKey(), "paid but not found created log"));
            } else {
                //如果pay没来，说明支付超时
                ctx.output(orderTimeoutTag, new OrderResult(ctx.getCurrentKey(), "paid timeout"));
            }
            isPaidState.clear();
            isCreatedState.clear();
            tsTimestampState.clear();
        }

        @Override
        public void processElement(OrderEvent value, Context ctx, Collector<OrderResult> out) throws Exception {
            Boolean isPaid = isPaidState.value();
            Boolean isCreated = isCreatedState.value();
            Long ts = tsTimestampState.value();

            if ("create".equals(value.getEventType())) {
                //1.判断是否支付过
                if (isPaid) {
                    //1.1如果已经正常支付，输出正常匹配的结果
                    out.collect(new OrderResult(value.getOrderId(), "paid successfully"));
                    //清空状态,删除定时器
                    isPaidState.clear();
                    isCreatedState.clear();
                    tsTimestampState.clear();
                    ctx.timerService().deleteEventTimeTimer(ts);
                } else {
                    Long timerTs = (value.getTimestamp() + 15 * 60) * 1000L;
                    ctx.timerService().registerEventTimeTimer(timerTs);
                    tsTimestampState.update(timerTs);
                    isCreatedState.update(true);
                }
            } else if ("pay".equals(value.getEventType())) {
                //2.判断是否由下单事件
                if (isCreated) {
                    //2.1 已经有过下单事件,要继续判断支付的时间戳是否超过15分钟
                    if (value.getTimestamp() * 1000L < ts) {
                        //在15分钟内没有超时，正常输出
                        out.collect(new OrderResult(value.getOrderId(), "paid successfully"));
                    } else {
                        //已经超时，输出侧输出流报警
                        ctx.output(orderTimeoutTag, new OrderResult(value.getOrderId(), "paid but already timeout " + value.getTimestamp()));
                    }
                    //清空状态,删除定时器
                    isPaidState.clear();
                    isCreatedState.clear();
                    tsTimestampState.clear();
                    ctx.timerService().deleteEventTimeTimer(ts);
                } else {
                    //2.2没有下单事件,乱序，注册一个定时器，等待下单事件
                    Long timerTs = value.getTimestamp() * 1000L;
                    ctx.timerService().registerEventTimeTimer(timerTs);
                    tsTimestampState.update(timerTs);
                    isPaidState.update(true);
                }
            }
        }
    }

}
